package cn.itcast.tags.models.ml

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.{MLModelTools, TagTools}
import org.apache.spark.ml.clustering.KMeansModel
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.{Column, DataFrame}

class PsmModel extends AbstractModel("消费敏感度",ModelType.ML){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    import businessDF.sparkSession.implicits._
    //root
    // |-- memberid: string (nullable = true)
    // |-- ordersn: string (nullable = true)
    // |-- orderamount: string (nullable = true)
    // |-- couponcodevalue: string (nullable = true)

    /*TODO 1.依据业务数据（订单数据）计算每个用户PSM值
           psm = 优惠订单占比 + 平均优惠金额占比 + 优惠总金额占比
                    tdonr         adar          tdar
           tdonr 优惠订单占比(优惠订单数 / 订单总数)
           adar 平均优惠金额占比(平均优惠金额 / 平均每单应收金额)
           tdar 优惠总金额占比(优惠总金额 / 订单总金额)
     */
    // 计算指标
    //ra: receivableAmount 应收金额
    val raColumn: Column = ($"orderamount" + $"couponcodevalue").as("ra")
    //da: discountAmount 优惠金额
    val daColumn: Column = $"couponcodevalue".cast(DataTypes.createDecimalType(10,2)).as("da")
    //pa: practicalAmount 实收金额
    val paColumn: Column = $"orderamount".cast(DataTypes.createDecimalType(10,2)).as("pa")
    //state: 订单状态，此订单是否是优惠订单，0表示非优惠订单，1表示优惠订单
    val stateColumn: Column = when($"couponcodevalue" === 0.0, 0).otherwise(1).as("state")
    //tdon 优惠订单数
    val tdonColumn: Column = sum($"state").as("tdon")
    //ton 总订单总数
    val tonColumn: Column = count($"state").as("ton")
    //tda 优惠总金额
    val tdaColumn: Column = sum($"da").as("tda")
    //tra 应收总金额
    val traColumn: Column = sum($"ra").as( "tra")
    /*
    tdonr 优惠订单占比(优惠订单数 / 订单总数)
    tdar 优惠总金额占比(优惠总金额 / 订单总金额)
    adar 平均优惠金额占比(平均优惠金额 / 平均每单应收金额)
    */
    val tdonrColumn: Column = ($"tdon" / $"ton").as("tdonr")
    val tdarColumn: Column = ($"tda" / $"tra").as("tdar")
    val adarColumn: Column = (
      ($"tda" / $"tdon") / ($"tra" / $"ton")
      ).as("adar")
    val psmColumn: Column = ($"tdonr" + $"tdar" + $"adar").as("psm")


    // 1. 业务数据（订单数据）计算PSM值
    val psmDF: DataFrame = businessDF
      // 确定每个订单ra、da、pa及是否为优惠订单
      .select(
        $"memberid".as("userId"), //
        raColumn, daColumn, paColumn, stateColumn //应收金额，优惠金额，实收金额，订单状态
      )
      // 按照userId分组，聚合统计：订单总数和订单总额
      .groupBy($"userId")
      .agg(
        tonColumn, tdonColumn, traColumn, tdaColumn//总订单总数,优惠订单数,应收总金额,优惠总金额
      )
      // 计算优惠订单占比、优惠总金额占比、adar
      .select(
        $"userId", tdonrColumn, tdarColumn, adarColumn//优惠订单占比,优惠总金额占比,平均优惠金额占比
      )
      // 计算PSM值
      .select($"*", psmColumn)
      .select(
        $"*", //
        when($"psm".isNull, 0.00000001)
          .otherwise($"psm").as("psm_score")
      )
    //root
    // |-- userId: string (nullable = true)
    // |-- tdonr: double (nullable = true)
    // |-- tdar: double (nullable = true)
    // |-- adar: double (nullable = true)
    // |-- psm: double (nullable = true)
    // |-- psm_score: double (nullable = true)

    val psmFeaturesDF: DataFrame = new VectorAssembler()
      .setInputCols(Array("psm_score"))
      .setOutputCol("features")
      .transform(psmDF)

    /*
     TODO:2、使用KMeans算法训练模型
    */
    val kMeansModel: KMeansModel = MLModelTools.loadModel(psmFeaturesDF, "psm",this.getClass).asInstanceOf[KMeansModel]

    /*
    * TODO:3、使用模型打标签
    * */
    val predictionDF: DataFrame = kMeansModel.transform(psmFeaturesDF)
    predictionDF.show(10,false)
    val modelDF: DataFrame = TagTools.kmeansMatchTag(kMeansModel, predictionDF, tagDF, "psm")


    modelDF
  }
}

object PsmModel{
  def main(args: Array[String]): Unit = {
    val model = new PsmModel
    model.executeModel(372L,false)
  }
}